package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.http.client.fluent.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author 张明诚
 * @date 2016-11-18 10:39:47
 * @version V1.0
 * @desc
 */
public class DeduplicationFilterForGroup implements FilterInterface {

	private final Logger LOG = LoggerFactory.getLogger(DeduplicationFilterForGroup.class);
	private String[] columnNames;
	private REFieldType[] types;
	private String docSchemaName;
	private Map<String, String> userDesc;
	private List<Object[]> datas;
	private final String[] urls;
	private final List<Object> deduplicationFields;

	public DeduplicationFilterForGroup(Configuration configuration) {
		this.urls = configuration.getStringArray(Constants.DEDUPLICATION_URL);
		this.deduplicationFields = configuration.getList(Constants.DEDUPLICATION_FIELDS);
	}

	@Override
	public REMessage filter(REMessage message) throws Exception {
		columnNames = message.colNames;
		types = message.types;
		docSchemaName = message.docSchemaName;
		userDesc = message.userDesc;
		datas = message.data;

		if (datas.isEmpty()) {
			return message;
		}

		String deduplicationValues = getDeduplicationValues();
		if (deduplicationValues.isEmpty()) {
			throw new Exception("deduplication values is empty!");
		} else {
			LOG.debug("deduplication values: \n{}", deduplicationValues);
		}
		String response = deduplicationService(deduplicationValues);
		LOG.debug("deduplication result:\n{}", response);
		String[] split = response.split("\n");
		List<String> flags = Arrays.asList(split);
		deleteDuplicationDatas(flags);
		REMessage deduplicationMessage
				= new REMessage(columnNames, types, datas, docSchemaName, userDesc);
		return deduplicationMessage;
	}

	public String getDeduplicationValues() {
		StringBuilder sb = new StringBuilder();
		List<String> columnList = Arrays.asList(columnNames);
		for (Object[] data : this.datas) {
			boolean isFirst = true;
			for (Iterator<Object> it = deduplicationFields.iterator(); it.hasNext();) {
				String distinctField = (String) it.next();
				int index = columnList.indexOf(distinctField);
				if (index < 0) {
					LOG.info(distinctField);
					LOG.error("the distinct field: {} does not exists!", distinctField);
				}
				Object distinctFieldValue = data[index];
				if (distinctFieldValue == null || distinctFieldValue.toString().isEmpty()) {
					LOG.info("columnames:{}", Arrays.toString(columnNames));
					LOG.info("data:{}", Arrays.toString(data));
					LOG.error("the value of distinct field: {} is null or empty!", distinctField);
				}
				if (!isFirst) {
					sb.append("^");
				}
				sb.append(distinctFieldValue);
				isFirst = false;
			}
			sb.append("\n");
		}
		return sb.toString();
	}

	public void deleteDuplicationDatas(List<String> flags) {
		List<Object[]> newMessage = new ArrayList<>();
		for (int i = 0; i < flags.size(); i++) {
			if (flags.get(i).equals("1")) {
				newMessage.add(datas.get(i));
			}
		}
		this.datas = newMessage;
	}

	private String deduplicationService(String content) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("deduplication content:{}", content);
		}
		String response = null;
		String url;
		int index;
		while (true) {
			index = new Random().nextInt(urls.length);
			url = urls[index];
			try {
				response = Request.Post(url)
						.bodyByteArray(content.getBytes())
						.connectTimeout(1800000)
						.socketTimeout(1800000)
						.execute()
						.returnContent().asString();
			} catch (IOException ex) {
				LOG.warn("request url {} error! \n"
						+ ExceptionUtils.getFullStackTrace(ex), url);
			}
			if (response == null || response.trim().isEmpty()) {
				LOG.error("response content is null，please check the url:{}", url);
			} else {
				break;
			}
			try {
				Thread.sleep(3000);
			} catch (InterruptedException ex) {
			}
		}
		return response;
	}

}
